package redemo;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 编写的时候尽量抽象，封装
 * 所以能传接口传接口
 */
class DemoSource {
    private volatile boolean flag = true; // 只需要保证可见性
    private final AtomicInteger id = new AtomicInteger(1);
    private BlockingQueue<String> blockingQueue = null;

    public DemoSource(BlockingQueue<String> queue) {
        this.blockingQueue = queue;
        System.out.println(queue.getClass().getName());
    }

    public void consume() throws InterruptedException {
        System.out.println("消费者开始消费");
        System.out.println();
        System.out.println();
        String res = null;
        while (flag) {
            res = blockingQueue.poll(2, TimeUnit.SECONDS);
            if (null == res || res.equalsIgnoreCase("")) { // 没有能够取的了
                System.out.println("消费者结束");
                return;
            }
            System.out.println("\t消费者取来商品" + res);
        }
        System.out.println();
        System.out.println();
        System.out.println("无商品可取，消费者停止");
    }

    public void produce() throws InterruptedException {
        System.out.println("生产者开始生产");
        System.out.println();
        System.out.println();
        boolean success = false;
        while (flag) {
            success = blockingQueue.offer(id.incrementAndGet() + "", 2, TimeUnit.SECONDS);
            if (!success) {
                System.out.println("队列已满，生产失败");
            } else {
                System.out.println("队列有空，生产商品" + id.get());
            }

            try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
        }
        System.out.println();
        System.out.println();
        System.out.println("主线要求停止生产，生产结束");
    }

    public void stop() {
        flag = false;
    }
}

/**
 * volatile/CAS/BlockingQueue/线程交互/原子引用
 * 阻塞队列最方便的就是生产者消费者模型
 * 能够由主线程控制生产消费行为的进行
 */
public class BlockingQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        DemoSource source = new DemoSource(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            try {
                source.produce();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        new Thread(() -> {
            try {
                source.consume();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
        source.stop();
        System.out.println("停止");
    }
}
